package main;

//import java.io.IOException;
//import java.text.DecimalFormat;
//import java.text.SimpleDateFormat;
//import java.util.ArrayList;
//import java.util.Date;
//import java.util.HashMap;
//import java.util.List;
//import java.util.Map;
//
//import org.apache.hadoop.conf.Configuration;
//import org.apache.hadoop.filecache.DistributedCache;
//import org.apache.hadoop.fs.FileSystem;
//import org.apache.hadoop.fs.Path;
//import org.apache.hadoop.io.LongWritable;
//import org.apache.hadoop.io.Text;
//import org.apache.hadoop.mapreduce.InputSplit;
//import org.apache.hadoop.mapreduce.Job;
//import org.apache.hadoop.mapreduce.Mapper;
//import org.apache.hadoop.mapreduce.Reducer;
//import org.apache.hadoop.mapreduce.Mapper.Context;
//import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
//import org.apache.hadoop.mapreduce.lib.input.FileSplit;
//import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
//import org.apache.hadoop.util.GenericOptionsParser;
//
//import tools.Dict;
//
//import com.bj58.datav3.format.track.TrackInputFormat;
//import com.bj58.datav3.model.TrackInfo;
//import com.bj58.datav3.tool.track.MyDateUtil;
//import com.bj58.datav3.tool.track.TrackSplitUtils;
//
///**
// *
//*
//* 类描述：月度UV统计
//* 创建人：caods
//* 创建时间：2015年6月4日 下午6:23:45
//* @version
//*
///
public class NetFlowMauJob {
//	public static class M1 extends Mapper<LongWritable, Text, Text, Text> {
//		TrackSplitUtils trackSplit = TrackSplitUtils.getInstance();
//		TrackInfo trackinfo = null;
//		protected void setup(Context context) throws IOException, InterruptedException{
//			Configuration conf=context.getConfiguration();
//			Path[] paths = DistributedCache.getLocalCacheFiles(conf);
//			Dict.marketSpmList=CatchFileTools.getListByCache(paths, "pcm_source_spm_list", "utf-8");
//			Dict.msourceLMMap=CatchFileTools.getMapByCache(paths, "m_source_lm_dict", "utf-8", "\t");
//			Dict.msourceQDMap=CatchFileTools.getMapByCache(paths, "m_source_qd_dict", "utf-8", "\t");
//			Dict.pcsourceLMList=CatchFileTools.getListByCache(paths, "m_source_lm_dict", "utf-8");
//			Dict.pcsourceList=CatchFileTools.getListByCache(paths, "pc_source_dict", "utf-8");
//			Dict.cityIdToNameMap=CatchFileTools.getCityCateIdMapByCache(paths, "cmc_diplocal", "utf-8", "\t","city");
//			Dict.cateIdToNameMap=CatchFileTools.getCityCateIdMapByCache(paths, "cmc_dispcategory", "utf-8", "\t", "cate");
//			Dict.cityIdPathMap=CatchFileTools.getCityCateIdMapByCache(paths, "cmc_diplocal", "utf-8", "\t","cityidPath");
//			Dict.cateIdPathMap=CatchFileTools.getCityCateIdMapByCache(paths, "cmc_dispcategory", "utf-8", "\t", "cateidPath");
//			Dict.businessMap=CatchFileTools.getCityCateIdMapByCache(paths, "dict_disp_cate", "utf-8", "\t", "bussiness");
//
//		}
//		public void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
//			InputSplit inputSplit = context.getInputSplit();
//			String pathName = ((FileSplit) inputSplit).getPath().getParent().getParent().getName();
//			String statDate = ((FileSplit) inputSplit).getPath().getParent().getName();// yyyyMMdd
//			String month = statDate.substring(0, 6);
//			Configuration conf=context.getConfiguration();
//			String platform=conf.get("plat");
//			String flag = conf.get("flag");
//			List<List<String>> listLines = trackSplit.getSplitVisit(value);
////			if(flag.equals("cate")){
////				Map<String, String> keyMap = MauKeyEntry.getCateKeyMap(listLines, platform, month, cateIdToYWMap,cateIdToNameMap,cateIdPathMap);
////				 for (Map.Entry<String, String> entry : keyMap.entrySet()) {
////					 context.write(new Text(entry.getValue()), new Text(entry.getKey()));
////				 }
////			}else if(flag.equals("source")){
////				Map<String, String> keyMap = MauKeyEntry.getSourceKeyMap(listLines, platform, month, cateIdToYWMap, pcsourceList, msourceLMMap, msourceQDMap,spmList,lmList);
////				 for (Map.Entry<String, String> entry : keyMap.entrySet()) {
////					 context.write(new Text(entry.getValue()), new Text(entry.getKey()));
////				 }
////			}else if(flag.equals("catesource")){
//				Map<String, String> keyMap = MauKeyEntry.getSourceCateKeyMap(listLines, platform, month,Dict.cityIdToNameMap, Dict.businessMap,Dict.cateIdToNameMap,Dict.cateIdPathMap, Dict.pcsourceList, Dict.msourceLMMap, Dict.msourceQDMap,Dict.marketSpmList,Dict.pcsourceLMList);
//				for (Map.Entry<String, String> entry : keyMap.entrySet()) {
//					 context.write(new Text(entry.getValue()), new Text(entry.getKey()));
//				 }
////			}
//
//		}
//	}
//	public static class R1 extends Reducer<Text, Text, Text, Text>{
//		public void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException {
//			Map<String, String> keyMap = new HashMap<String, String>();
//			for (Text text : values) {
//				keyMap.put(text.toString().replace("\t", "###"), "");
//			}
//			for (Map.Entry<String, String> entry : keyMap.entrySet()) {
//				context.write(new Text(entry.getKey()) , key);
//			}
//		}
//	}
//
//	public static class M2 extends Mapper<LongWritable, Text, Text, Text> {
//		public void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
//			InputSplit inputSplit = context.getInputSplit();
//			String pathName = ((FileSplit) inputSplit).getPath().getParent().getParent().getName();
//			String time = ((FileSplit) inputSplit).getPath().getParent().getName();//yyyyMMdd
//			String newVal=value.toString();
//			context.write(new Text(newVal.split("\t",-1)[0]) , new Text(newVal.split("\t",-1)[1]));
//		}
//	}
//
//	public static class R2 extends Reducer<Text, Text, Text, Text>{
//		public void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException {
//			int uv = 0;
//			for (Text text : values) {
//				uv+=1;
//			}
//			context.write(new Text(key.toString().replace("###", "\t")) , new Text(uv+""));
//		}
//	}
//
//	public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException
//	{
//		Configuration conf = new Configuration();
//		String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs();
//		/**********************************日期******************************/
//		Date currentTime = new Date();
//		SimpleDateFormat smdf = new SimpleDateFormat("yyyyMMdd");
//		String yestadyTimeStr = smdf.format(currentTime.getTime() - 1000 * 60 * 60 * 24);
//		String startDate = "";
//		String endDate = "";
//		String[] platArr = null;//"pc,m".split(",");
//		String[] flagArr = null;// "cate,source,catesource".split(",");
//		if(otherArgs.length>0){
//			startDate = otherArgs[0];
//			endDate = otherArgs[1];
//			platArr = new String[]{otherArgs[2]};
//			flagArr = new String[]{otherArgs[3]};
//		}else{
//			startDate = yestadyTimeStr;
//			endDate = yestadyTimeStr;
//		}
//		int returnCode = 126;
//		FileSystem.get(conf).delete(new Path("/home/hdp_lbg_ecdata_dw/middata/caods/netflow/mau/job1/"), true);
//		for (String plat : platArr) {
//			for (String flag : flagArr) {
//				conf.set("plat", plat);
//				conf.set("flag", flag);
//				Job job = new Job(conf, "NetFlowMauJob");
//				job.setInputFormatClass(TrackInputFormat.class);
//				job.setJarByClass(NetFlowMauJob.class);
//				job.setMapperClass(M1.class);
//				job.setReducerClass(R1.class);
//				job.setNumReduceTasks(100);
//				job.setOutputKeyClass(Text.class);
//				job.setOutputValueClass(Text.class);
//				job.setMapOutputKeyClass(Text.class);
//				job.setMapOutputValueClass(Text.class);
//				//****************添加缓存**********************
//				job=CatchFileTools.addCache(job, conf, "/home/hdp_lbg_ecdata_dw/warehouse/hdp_lbg_ecdata_dw_defaultdb/dict_source_type_pc", "pc_source_dict");
//				job=CatchFileTools.addCache(job, conf, "/home/hdp_lbg_ecdata_dw/warehouse/hdp_lbg_ecdata_dw_defaultdb/dict_m_source_qd","m_source_qd_dict");
//				job=CatchFileTools.addCache(job, conf, "/home/hdp_lbg_ecdata_dw/warehouse/hdp_lbg_ecdata_dw_defaultdb/dict_m_source_lm","m_source_lm_dict");
//				job=CatchFileTools.addCache(job, conf, "/home/hdp_lbg_ecdata_dw/warehouse/hdp_lbg_ecdata_dw_defaultdb/dict_m_market_spm","pcm_source_spm_list");
//				job=CatchFileTools.addCache(job, conf, "/home/hdp_lbg_ecdata_dw/warehouse/hdp_lbg_ecdata_dw_defaultdb/dict_disp_cate", "dict_disp_cate");
//				job=CatchFileTools.addCache(job, conf, "/home/hdp_lbg_ecdata_dw/warehouse/hdp_lbg_ecdata_dw_defaultdb/dict_disp_city", "dict_disp_city");
//				job=CatchFileTools.addCache(job, conf, "/home/hdp_58_common/rawdata/cmc/dict/cmc_dispcategory/"+yestadyTimeStr, "cmc_dispcategory");
//				job=CatchFileTools.addCache(job, conf, "/home/hdp_58_common/rawdata/cmc/dict/cmc_displocal/"+yestadyTimeStr, "cmc_diplocal");
//				//输出路径
//				String baseOutPath_1 = "/home/hdp_lbg_ecdata_dw/middata/caods/netflow/mau/job1/"+flag+"/"+startDate.substring(0, 6)+"/"+plat;
//				for (String runDate : MyDateUtil.getDateList(startDate, endDate, 0)) {
//					if(plat.equals("pc")){
//						FileInputFormat.addInputPath(job, new Path("/home/hdp_58_common/resultdata/all_site_user_action_etl/pc.track.58.com/" + runDate));
//						System.out.println("############"+plat+"	第一次输入路径为#########/home/hdp_58_common/resultdata/all_site_user_action_etl/pc.track.58.com/"+runDate);
//					}else{
//						FileInputFormat.addInputPath(job, new Path("/home/hdp_58_common/resultdata/all_site_user_action_etl/m.track.58.com/" + runDate));
//						System.out.println("############"+plat+"	第一次输入路径为#########/home/hdp_58_common/resultdata/all_site_user_action_etl/m.track.58.com/"+runDate);
//						//MAU---------------------------------微信的日志是20140901开始的------------------------
//						if(Integer.parseInt(runDate.substring(0, 6))>201408){
//							FileInputFormat.addInputPath(job, new Path("/home/hdp_58_common/resultdata/all_site_user_action_etl/weixin.track.58.com/" + runDate));
//							System.out.println("############"+plat+"	第一次输入路径为#########/home/hdp_58_common/resultdata/all_site_user_action_etl/weixin.track.58.com/"+runDate);
//						}
//					}
//				}
//
//				FileOutputFormat.setOutputPath(job, new Path(baseOutPath_1));
//				System.out.println("############"+plat+"	第一次输出路径为#########"+baseOutPath_1);
//				job.waitForCompletion(true);
//
//				Job job2 = new Job(conf, "NetFlowMauJob_2");
//				job2.setJarByClass(NetFlowMauJob.class);
//				job2.setMapperClass(M2.class);
//				job2.setReducerClass(R2.class);
//				job2.setNumReduceTasks(30);
//				job2.setOutputKeyClass(Text.class);
//				job2.setOutputValueClass(Text.class);
//				job2.setMapOutputKeyClass(Text.class);
//				job2.setMapOutputValueClass(Text.class);
//
//				String baseOutPath_2 = "/home/hdp_lbg_ecdata_dw/resultdata/caods/netflow/mau/job2/"+flag+"/"+startDate.substring(0, 6)+"/"+plat;
//				FileInputFormat.addInputPath(job2, new Path(baseOutPath_1));
//				FileSystem.get(conf).delete(new Path(baseOutPath_2), true);
//				FileOutputFormat.setOutputPath(job2, new Path(baseOutPath_2));
//				System.out.println("############"+plat+"	第二次输入路径为#########"+baseOutPath_1);
//				System.out.println("############"+plat+"	第二次输出路径为#########"+baseOutPath_2);
//				returnCode = job2.waitForCompletion(true) ? 0 : 1;
//				if (returnCode != 0) {
//					System.exit(returnCode);
//				}
//			}
//		}
//		System.exit(returnCode);
//	}
//
}
